package com.spider.moudle.factory;

import com.google.common.collect.Lists;
import com.spider.moudle.util.PropertiesUtil;
import com.spider.moudle.util.SettingsUtil;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * Created by Linbo Hu on 2017/12/4.
 * kafka消费者工厂
 */
public class KafkaConsumerFactory {
    private static Properties comsumer_properties = new Properties();

    static {
        comsumer_properties.put("zookeeper.connect", PropertiesUtil.getVal("kafka.zookeeper.path"));
        comsumer_properties.put("zookeeper.session.timeout.ms", "40000");
        comsumer_properties.put("zookeeper.sync.time.ms", "2000");
        comsumer_properties.put("auto.commit.interval.ms", "10000");
        comsumer_properties.put("rebalance.backoff.ms", "12000");
        comsumer_properties.put("rebalance.max.retries", "5");
        comsumer_properties.put("auto.offset.reset", "smallest");
        // max.poll.records 每次拉取消息的条数
        comsumer_properties.put("serializer.class", "kafka.serializer.StringEncoder");
        comsumer_properties.put("group.id", PropertiesUtil.getVal("kafka.task.group.id"));
    }

    public static ConsumerIterator<String, String> build(String topic) {
        ConsumerConfig consumerConfig = new ConsumerConfig(comsumer_properties);
        ConsumerConnector consumer = kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);
        Map<String, Integer> topicCountMap = new HashMap<>();
        topicCountMap.put(topic, 1);
        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());
        Map<String, List<KafkaStream<String, String>>> consumerMap =
                consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
        List<ConsumerIterator<String, String>> list = Lists.newArrayList();
        KafkaStream<String, String> stream = consumerMap.get(topic).get(0);
        return stream.iterator();
    }
}
